package gnssio

import (
	"dgo/goutils/dnet"
	"dgo/goutils/utils"
	"errors"
	"fmt"
	"math"
	"net"
	"strconv"
	"strings"
	"sync/atomic"
	"time"
)

type CommTcpSvr struct {
	id string
	//onlinelocker      *sync.RWMutex
	objDNA              uint32
	NetInfo             *dnet.DNetInfo
	MonitorSendTaskIO   utils.MonitorSpeedNTimeRec
	MonitorSendTaskSize utils.MonitorSpeedSizeRec
	flag                byte
	addr                string
	netListen           net.Listener
	isTerminated        bool
	startFlag           int32
	BlockRecvSize       uint16
	listeners           *utils.SafeList
	timeout_sec_kick    int                 // 如果>0， 则进行心跳检测
	maxCount            int32               // 最大连接数, 超过最大连接数据会随机删除一个连接， 0：不限制
	onlineConnections   utils.GSyncMapGroup // 在线列表
	onlineCount         int32               // 在线数量
	MonitorFlag         int                 // 侦测类型 0:不跟踪, 1:跟踪详细
	UseQueueSend        bool                // 使用队列进行发送, 如果是队列. 必须保证Buf 不能被重复修改, 队列发送, 是异步进行的
	OnRecvBufEvent      func(sender *CommTcpSvrClientConnection, buf []byte)
	OnDisEvent          func(sender *CommTcpSvrClientConnection)
	OnConnEvent         func(sender *CommTcpSvrClientConnection)
}

func NewCommTcpSvr(id string, addr string, maxCount int32, timeout_sec int) *CommTcpSvr {
	rval := &CommTcpSvr{
		flag:             0,
		objDNA:           utils.NewObjectDNA(),
		NetInfo:          &dnet.DNetInfo{},
		startFlag:        0,
		id:               id,
		addr:             addr,
		BlockRecvSize:    dnet.RECV_BLOCK_SIZE,
		maxCount:         maxCount,
		timeout_sec_kick: timeout_sec,
		listeners:        utils.NewSafeList(),
		//onlinelocker:      new(sync.RWMutex),
		// onlineConnections: make(map[uint32]*CommTcpSvrClientConnection),
	}
	utils.Monitor.AddMonitorRec(fmt.Sprintf("%d_task_io", rval.objDNA), &rval.MonitorSendTaskIO)
	utils.Monitor.AddMonitorRec(fmt.Sprintf("%d_task_size", rval.objDNA), &rval.MonitorSendTaskSize)
	return rval
}

/*
**

	tcpsvr://ip:port[/maxcount[/timeout_sec/use_queue]]
*/
func NewCommTcpSvrFromCmd(cmdstr string) *CommTcpSvr {
	if len(cmdstr) < 9 {
		return nil
	}

	ok, cmdstr := utils.CheckTrimPrefix(cmdstr, "tcpsvr://")
	if !ok {
		return nil
	}

	strs := strings.SplitN(cmdstr, "/", 2)
	addr := strs[0]
	maxcnt := int32(0)
	timeout_sec := 0
	use_queue := false
	if len(strs) == 2 {
		// maxcount[/timeout_sec]
		cmdstr = strs[1]
		strs = strings.SplitN(cmdstr, "/", 2)
		i64, _ := strconv.ParseInt(strs[0], 10, 8)
		maxcnt = int32(i64)
		if len(strs) == 2 {
			cmdstr = strs[1]
			strs = strings.SplitN(cmdstr, "/", 2)
			i64, _ := strconv.ParseInt(strs[0], 10, 8)
			timeout_sec = int(i64)
			if len(strs) == 2 {
				use_queue = utils.TryStrToBool(strs[1], false)
			}
		}
	}

	rval := NewCommTcpSvr(addr, addr, maxcnt, timeout_sec)
	rval.UseQueueSend = use_queue
	return rval
}

/*
**

	使用队列
	tcpsvr2://ip:port[/maxcount[/timeout_sec/use_queue]]
*/
func NewCommASyncTcpSvrFromCmd(cmdstr string) *CommTcpSvr {
	if len(cmdstr) < 9 {
		return nil
	}

	ok, cmdstr := utils.CheckTrimPrefix(cmdstr, "tcpsvr2://")
	if !ok {
		return nil
	}

	strs := strings.SplitN(cmdstr, "/", 2)
	addr := strs[0]
	maxcnt := int32(0)
	timeout_sec := 0
	if len(strs) == 2 {
		// maxcount[/timeout_sec]
		cmdstr = strs[1]
		strs = strings.SplitN(cmdstr, "/", 2)
		i64, _ := strconv.ParseInt(strs[0], 10, 8)
		maxcnt = int32(i64)
		if len(strs) == 2 {
			i64, _ := strconv.ParseInt(strs[1], 10, 8)
			timeout_sec = int(i64)
		}
	}

	rval := NewCommTcpSvr(addr, addr, maxcnt, timeout_sec)
	rval.UseQueueSend = true
	return rval
}

func CreateTcpSvrCommFromCmdLineFunc(args ...interface{}) (rval interface{}, err error) {
	if len(args) < 1 {
		return nil, errors.New("至少要传入一个参数")
	}
	cmdStr := args[0].(string)
	if len(cmdStr) == 0 {
		return nil, errors.New("命令参数非法")
	}
	return NewCommTcpSvrFromCmd(cmdStr), nil
}

func CreateASyncTcpSvrCommFromCmdLineFunc(args ...interface{}) (rval interface{}, err error) {
	if len(args) < 1 {
		return nil, errors.New("至少要传入一个参数")
	}
	cmdStr := args[0].(string)
	if len(cmdStr) == 0 {
		return nil, errors.New("命令参数非法")
	}
	return NewCommASyncTcpSvrFromCmd(cmdStr), nil
}

func (this *CommTcpSvr) RangeOnlineList(fn func(conn *CommTcpSvrClientConnection) bool) (cnt int) {
	// 踢掉较早的连接
	onlines := this.onlineConnections.ListSorted(func(iItm, jItm interface{}) bool {
		iConn := iItm.(*CommTcpSvrClientConnection)
		jConn := jItm.(*CommTcpSvrClientConnection)
		return iConn.handle < jConn.handle
	})

	for i := 0; i < len(onlines); i++ {
		if !fn(onlines[i].(*CommTcpSvrClientConnection)) {
			break
		}
		cnt++
	}
	return
}

func (this *CommTcpSvr) innerCheckKickOut() {
	if this.maxCount == 0 {
		return
	}

	if this.maxCount > this.onlineCount {
		return
	}

	// 踢掉较早的连接
	onlines := this.onlineConnections.ListSorted(func(iItm, jItm interface{}) bool {
		iConn := iItm.(*CommTcpSvrClientConnection)
		jConn := jItm.(*CommTcpSvrClientConnection)
		return iConn.connecttime.Before(jConn.connecttime)
	})

	cnt := len(onlines)
	for {
		for i := 0; i < len(onlines); i++ {
			if cnt < int(this.maxCount) {
				break
			}

			conn := onlines[i].(*CommTcpSvrClientConnection)
			conn.RequestDisconnect()
			cnt--
		}
	}
}

func (this *CommTcpSvr) innerStartListen() {
	utils.Log.Infof("%s侦听地址:%s", this.id, this.addr)
	netListen, err := net.Listen("tcp", this.addr)
	if err != nil {
		utils.Log.Errf("%s侦听地址:%s 出现异常:%v", this.id, this.addr, err)
		return
	}

	this.netListen = netListen

	defer netListen.Close()

	this.isTerminated = false

	utils.Log.Infof("%s等待连接", this.id)
	for {
		_conn, err := netListen.Accept()
		if err != nil {
			if this.isTerminated {
				break
			} else {
				continue
			}
		} else {
			// 检测是否要T掉之前的连接
			this.innerCheckKickOut()
		}

		if this.isTerminated {
			utils.Log.Infof("%s，即将关闭，不处理新的接入", this.id)
			_conn.Close()
			break
		}
		ctx := NewCommTcpSvrClientConnection(_conn, this)
		ctx.connecttime = time.Now()
		this.addOnline(ctx)
		ctx.Start()
	}
}

/*
*

	根据Handle查找连接(在线)
*/
func (self *CommTcpSvr) FindConnByHandle(handle uint32) (ctx *CommTcpSvrClientConnection) {
	obj := self.onlineConnections.Get(handle, nil)
	if obj == nil {
		return nil
	}
	return obj.(*CommTcpSvrClientConnection)
}

func (self *CommTcpSvr) RequestDisByHandle(handle uint32, disReson string) bool {
	ctx := self.FindConnByHandle(handle)
	if ctx != nil {
		ctx.disReason = disReson
		ctx.RequestDisconnect()
		return true
	}
	return false
}

func (self *CommTcpSvr) addOnline(ctx *CommTcpSvrClientConnection) {
	_, loaded := self.onlineConnections.LoadOrStore(ctx.handle, ctx)
	if !loaded {
		atomic.AddInt32(&self.onlineCount, 1)
		utils.Log.Debugf("连接(%d:%s)进入, 当前在线:%d", ctx.handle, ctx.conn.RemoteAddr().String(), self.onlineCount)
	} else {
		utils.Log.WarnTypef("BUG", "连接(%d:%s)进入, handle重复!, 当前在线数:%d", ctx.handle, ctx.conn.RemoteAddr().String(), self.onlineCount)
	}
	if self.OnConnEvent != nil {
		self.OnConnEvent(ctx)
	}
}

func (self *CommTcpSvr) removeOnline(ctx *CommTcpSvrClientConnection) {
	if self.onlineConnections.Contain(ctx.handle) {
		self.onlineConnections.Remove(ctx.handle)
		utils.Log.Debugf("连接(%d:%s)断开, 当前在线:%d", ctx.handle, ctx.conn.RemoteAddr().String(), self.onlineCount)
		atomic.AddInt32(&self.onlineCount, -1)
	} else {
		utils.Log.WarnTypef("BUG", "连接(%d:%s)断开, 当前在线:%d, Onlines不包含Handle", ctx.handle, ctx.conn.RemoteAddr().String(), self.onlineCount)
		self.onlineConnections.Remove(ctx.handle)
	}

	ctx.distime = time.Now()

	if ctx.OnDisEvent != nil {
		ctx.OnDisEvent(ctx)
	}

	if self.OnDisEvent != nil {
		self.OnDisEvent(ctx)
	}
}

func (this *CommTcpSvr) onRecvBuffer(ctx *CommTcpSvrClientConnection, buf []byte, l int) {
	if this.OnRecvBufEvent != nil {
		this.OnRecvBufEvent(ctx, buf[:l])
	}
	this.listeners.ForEach(func(val interface{}) {
		val.(I_CommListener).OnReceive(this, buf, l)
	})
}

// 检测超时连接
func (this *CommTcpSvr) CheckKickOut(timeout_isec int) {
	sleeptime := time.Duration(math.Floor(float64(timeout_isec) / 2))
checkfor:
	for {
		if this.isTerminated {
			break checkfor
		}
		time.Sleep(time.Second * time.Duration(sleeptime))
		this.onlineConnections.Range(func(key, value interface{}) bool {
			ctx := value.(*CommTcpSvrClientConnection)
			ctx.checkKickOut(timeout_isec)
			return true
		})

	}
}

func (this *CommTcpSvr) DisconnectAll() {
	this.onlineConnections.Range(func(key, value interface{}) bool {
		ctx := value.(*CommTcpSvrClientConnection)
		ctx.RequestDisconnect()
		return true
	})
}

// =====I_COMM
func (this *CommTcpSvr) Start() {
	if !atomic.CompareAndSwapInt32(&this.startFlag, 0, 1) {
		return
	}

	this.isTerminated = false
	if this.timeout_sec_kick > 0 {
		utils.GoFunc2(func() {
			this.CheckKickOut(this.timeout_sec_kick)
		})
	}
	utils.GoFunc2(this.innerStartListen)
}

func (this *CommTcpSvr) StartSync() {
	if !atomic.CompareAndSwapInt32(&this.startFlag, 0, 1) {
		return
	}

	this.isTerminated = false
	if this.timeout_sec_kick > 0 {
		utils.GoFunc2(func() {
			this.CheckKickOut(this.timeout_sec_kick)
		})
	}
	this.innerStartListen()
}

func (this *CommTcpSvr) Stop() {
	if !atomic.CompareAndSwapInt32(&this.startFlag, 1, 0) {
		return
	}

	this.isTerminated = true
	this.DisconnectAll()
	this.netListen.Close()
}

func (this *CommTcpSvr) Send(buf []byte, l int) int {
	defer utils.PanicHandler()
	bufl := l
	if bufl == 0 {
		bufl = len(buf)
	}

	this.MonitorSendTaskSize.Delta(int64(bufl))

	this.onlineConnections.Range(func(key, value interface{}) bool {
		ctx := value.(*CommTcpSvrClientConnection)
		ctx.Send(buf, l)
		return true
	})

	if this.onlineCount > 0 {
		return l
	} else {
		return 0
	}
}

func (this *CommTcpSvr) AddListener(event I_CommListener) {
	if this.listeners.Exists(event) {
		utils.Log.Warn("I_CommListener已经存在请勿重复侦听")
		return
	}
	this.listeners.Add(event)
}

func (this *CommTcpSvr) RemoveListener(event I_CommListener) {
	this.listeners.Remove(event)
}
